package cn.org.intelli.zffserver.service;

import cn.org.intelli.zffserver.domain.JobQueue;
import cn.org.intelli.zffserver.entity.*;
import cn.org.intelli.zffserver.repository.*;
import cn.org.intelli.zffserver.util.HttpClient;
import cn.org.intelli.zffserver.util.Uid;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import dm.jdbc.util.StringUtil;
import org.apache.commons.lang3.StringUtils;
import org.beetl.core.GroupTemplate;
import org.beetl.core.Template;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.net.URLEncoder;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


@Service
public class JobService {
    /*  32 */   private static final Logger log = LoggerFactory.getLogger(cn.org.intelli.zffserver.service.JobService.class);
    private static Boolean queueExecuteStatus = Boolean.valueOf(false);
    ExecutorService executor = Executors.newFixedThreadPool(1);
    @Value("${spark.server}")
    private String sparkServer;
    @Value("${spark.restfulPort}")
    private String sparkRestfulPort;
    @Value("${spark.port}")
    private String sparkPort;
    @Value("${spark.submitServer}")
    private String sparkSubmitServer;
    @Value("${spark.jarDir}")
    private String jarDir;
    @Value("${spark.version}")
    private String sparkVersion;
    @Value("${spark.cluster.cores}")
    private Integer cores;
    @Autowired
    private HttpClient httpClient;
    @Autowired
    private JobRepository jobRepository;
    @Autowired
    private ModelRepository modelRepository;
    @Autowired
    private DeletedJobRepository deletedJobRepository;
    @Autowired
    private DataBaseRepository dataBaseRepository;
    @Autowired
    private DataBaseV2Repository dataBaseV2Repository;
    @Autowired
    private ExecutorJarRepository executorJarRepository;
    @Autowired
    private GroupTemplate groupTemplate;

    public synchronized void jobExecute(JobEntity job) {
        try {
            log.info("@JobExecute: 开始执行任务，jobId为{}", job.getId());
            Boolean resourceEnough = checkSparkServerResource(Integer.valueOf(1), Long.valueOf(1L));
            if (!resourceEnough.booleanValue()) {
                log.info("@JobExecute: 当前Spark资源已满，将任务{}添加至等待队列中", job.getId());
                JobQueue.addJob(job);
                if (!queueExecuteStatus.booleanValue()) {
                    queueExecute();
                    queueExecuteStatus = Boolean.valueOf(true);
                }
            } else {
                log.info("@JobExecute: 开始创建Spark任务{}", job.getId());
                jobSubmit2Spark(job);
                log.info("@JobExecute: Spark任务{}提交成功", job.getId());
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("@JobExecute: error is {}", e.getMessage());
            job.setStatus("FAILED");
            job.setMessage(e.getMessage());
            this.jobRepository.save(job);
        }
    }


    private void createResultTable(ModelEntity model) throws Exception {
        /* 106 */
        String destinationDbId = model.getDestinationDbId();
        /* 107 */
        DataBaseV2Entity database = this.dataBaseV2Repository.findById(destinationDbId).get();
        /* 108 */
        String dbType = database.getDbType();

        /* 110 */
        switch (dbType) {
            case "dm":
                /* 112 */
                createResultTableInDm(model, database);
                break;
            case "clickhouse":
                /* 115 */
                createResultTableInClickhouse(model, database);
                break;
        }
    }


    private void createResultTableInDm(ModelEntity model, DataBaseV2Entity database) throws Exception {
        /* 127 */
        JSONObject dbInfo = JSONObject.parseObject(database.getDbInfo());

        /* 129 */
        String resultTableName = model.getModelCode() + "_RESULT";
        /* 130 */
        String checkSqlFormat = "SELECT COUNT(0) AS COUNT FROM ALL_TABLES WHERE TABLE_NAME = '%s' AND OWNER = '%s'";
        /* 131 */
        String checkSql = String.format(checkSqlFormat, resultTableName, dbInfo.getString("schemaName"));

        /* 133 */
        Class.forName(dbInfo.getString("driverName"));

        /* 135 */
        Connection conn = DriverManager.getConnection(dbInfo.getString("dbUrl"), dbInfo
/* 136 */.getString("dbUserName"), dbInfo.getString("dbPassword"));
        /* 137 */
        Statement smt = conn.createStatement();
        try {
            /* 139 */
            ResultSet rs1 = smt.executeQuery(checkSql);
            /* 140 */
            Integer checkNum = Integer.valueOf(0);
            /* 141 */
            if (rs1.next()) {
                /* 142 */
                checkNum = Integer.valueOf(rs1.getInt("COUNT"));
            }

            /* 145 */
            if (checkNum.intValue() == 1) {
                /* 146 */
                log.info("DM数据库中{}库下{}表已创建，不再重复创建", dbInfo.getString("schemaName"), resultTableName);
            } else {
                /* 148 */
                log.info("开始在DM数据库中创建{}.{}表", dbInfo.getString("schemaName"), resultTableName);
                /* 149 */
                StringBuffer createSql = new StringBuffer();
                /* 150 */
                createSql.append(
                        /* 151 */             String.format("CREATE TABLE %s.%s(\n", dbInfo.getString("schemaName"), resultTableName));
                /* 152 */
                if (StringUtil.isEmpty(model.getResultMeta())) {
                    /* 153 */
                    log.error("模型结果集元数据配置为空，任务执行失败");
                    /* 154 */
                    throw new Exception("模型结果集元数据配置为空，任务执行失败");
                }
                /* 156 */
                JSONArray resultMeta = JSON.parseArray(model.getResultMeta());
                /* 157 */
                for (int i = 0; i < resultMeta.size(); i++) {
                    /* 158 */
                    JSONObject atom = resultMeta.getJSONObject(i);
                    /* 159 */
                    createSql.append(String.format("%s %s,\n", atom
                            /* 160 */.getString("columnName"), atom.getString("columnType")));
                }
                /* 162 */
                createSql.append("JOBID VARCHAR(255)\n");
                /* 163 */
                createSql.append(") STORAGE(ON \"MAIN\", CLUSTERBTR)\n");
                /* 164 */
                log.info("@CreateResultTableInDm: create sql is {}", createSql);
                /* 165 */
                smt.execute(createSql.toString());
                /* 166 */
                log.info("目标表{}创建完成", resultTableName);
            }
            /* 168 */
        } catch (Exception e) {
            /* 169 */
            e.printStackTrace();
            /* 170 */
            throw new Exception(e);
        } finally {
            /* 172 */
            smt.close();
            /* 173 */
            conn.close();
        }
    }


    private void createResultTableInClickhouse(ModelEntity model, DataBaseV2Entity database) throws Exception {
        /* 184 */
        String resultTableName = model.getModelCode() + "_RESULT";

        /* 186 */
        JSONObject dbInfo = JSONObject.parseObject(database.getDbInfo());

        /* 188 */
        JSONArray resultMeta = JSON.parseArray(model.getResultMeta());
        /* 189 */
        StringBuffer createSql = new StringBuffer();

        /* 191 */
        String clusterName = dbInfo.getString("clusterName");

        /* 193 */
        if (StringUtils.isNotEmpty(clusterName)) {
            /* 194 */
            createSql.append(
                    /* 195 */           String.format("CREATE TABLE IF NOT EXISTS %s.%s ON cluster %s (\n", /* 196 */               dbInfo.getString("schemaName"), resultTableName, clusterName));
        } else {
            /* 198 */
            createSql.append(
                    /* 199 */           String.format("CREATE TABLE IF NOT EXISTS %s.%s (\n", /* 200 */               dbInfo.getString("schemaName"), resultTableName));
        }
        /* 202 */
        for (int i = 0; i < resultMeta.size(); i++) {
            /* 203 */
            JSONObject meta = resultMeta.getJSONObject(i);
            /* 204 */
            String columnName = meta.getString("columnName");
            /* 205 */
            String columnType = meta.getString("columnType");
            /* 206 */
            createSql.append(String.format("\"%s\" %s,\n", columnName, columnType));
        }
        /* 208 */
        createSql.append("\"JOBID\" String");

        /* 210 */
        String engineBeetl = dbInfo.getString("engine");

        /* 212 */
        JSONObject engineParam = new JSONObject();
        /* 213 */
        engineParam.put("schemaName", dbInfo.getString("schemaName"));
        /* 214 */
        engineParam.put("tableName", resultTableName);
        /* 215 */
        Template t = this.groupTemplate.getTemplate(engineBeetl);
        /* 216 */
        t.binding(engineParam);
        /* 217 */
        String engine = t.render();

        /* 219 */
        createSql.append(") ENGINE = " + engine + "\n");
        /* 220 */
        createSql.append("PARTITION BY (JOBID)\n");
        /* 221 */
        createSql.append("ORDER BY (JOBID)");
        /* 222 */
        log.info(createSql.toString());

        /* 224 */
        String createDataBaseSql = null;
        /* 225 */
        if (StringUtils.isNotEmpty(clusterName)) {

            /* 227 */
            createDataBaseSql = String.format("CREATE DATABASE IF NOT EXISTS \"%s\" ON cluster %s", /* 228 */             dbInfo.getString("schemaName"), clusterName);
        } else {
            /* 231 */
            createDataBaseSql = String.format("CREATE DATABASE IF NOT EXISTS \"%s\"", /* 232 */             dbInfo.getString("schemaName"));
        }
        /* 235 */
        System.out.println(createDataBaseSql);

        /* 237 */
        Class.forName(dbInfo.getString("driverName"));

        /* 239 */
        Connection conn = DriverManager.getConnection(dbInfo.getString("dbUrl"), dbInfo
/* 240 */.getString("dbUserName"), dbInfo.getString("dbPassword"));
        /* 241 */
        Statement smt = conn.createStatement();
        try {
            /* 243 */
            smt.execute(createDataBaseSql);
            /* 244 */
            smt.execute(createSql.toString());
            /* 245 */
        } catch (Exception e) {
            /* 246 */
            e.printStackTrace();
            /* 247 */
            throw new Exception(e);
        } finally {
            /* 249 */
            smt.close();
            /* 250 */
            conn.close();
        }
    }


    /* private synchronized void queueExecute() {
         log.info("@QueueExecute: 开启Job队列监听方法");
         this.executor.execute((Runnable) new Object(this));
         log.info("@QueueExecute: 当前队列中任务已执行完毕");
     }
 */
    private synchronized void queueExecute() {
        log.info("@QueueExecute: 开启Job队列监听方法");

        this.executor.execute(() -> {
            try {
                while (true) {
                    JobEntity nextJob = JobQueue.getTopJob();
                    if (nextJob == null) {
                        log.info("@QueueExecute: 队列为空，停止监听");
                        break;
                    }

                    Boolean resourceEnough = checkSparkServerResource(1, 1L);
                    if (resourceEnough) {
                        log.info("@QueueExecute: 资源充足，执行任务 {}", nextJob.getId());
                        jobSubmit2Spark(nextJob);
                    } else {
                        log.info("@QueueExecute: 资源不足，将任务 {} 重新放回队列", nextJob.getId());
                        JobQueue.addJob(nextJob);
                        // 等待一段时间后重试
                        Thread.sleep(10000);
                    }

                    // 添加小延迟，避免CPU过度占用
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                log.info("@QueueExecute: 队列监听被中断");
                Thread.currentThread().interrupt();
            } catch (Exception e) {
                log.error("@QueueExecute: 队列执行出错 - {}", e.getMessage(), e);
            } finally {
                queueExecuteStatus = false;
                log.info("@QueueExecute: 队列监听结束");
            }
        });

        log.info("@QueueExecute: 队列监听任务已启动");
    }


    private synchronized void jobSubmit2Spark(JobEntity job) throws Exception {
        ModelEntity model = this.modelRepository.findById(job.getModelId()).get();
        createResultTable(model);

        ExecutorJarEntity executorJar = this.executorJarRepository.findById(model.getExecutorId()).get();
        String jarFileName = executorJar.getFileName();
        String jarPath = this.jarDir + "/" + jarFileName;

        JSONObject sparkSubmitData = new JSONObject();

        sparkSubmitData.put("action", "CreateSubmissionRequest");
        sparkSubmitData.put("appResource", jarPath);
        sparkSubmitData.put("clientSparkVersion", this.sparkVersion);
        JSONObject environmentVariables = new JSONObject();
        environmentVariables.put("SPARK_ENV_LOADED", "1");
        sparkSubmitData.put("environmentVariables", environmentVariables);
        sparkSubmitData.put("mainClass", "entrance.Executer_Entrance");

        JSONObject sparkProperties = new JSONObject();
        sparkProperties.put("spark.driver.supervise", "false");
        sparkProperties.put("spark.app.name", job.getId());
        sparkProperties.put("spark.submit.deployMode", "cluster");
        sparkProperties.put("spark.master", this.sparkSubmitServer);
        sparkProperties.put("spark.executor.memory", "512m");
        sparkProperties.put("spark.executor.cores", "1");
        sparkProperties.put("spark.cores.max", "1");
        sparkProperties.put("spark.driver.memory", "512m");
        sparkProperties.put("spark.driver.cores", "1");
        sparkProperties.put("spark.speculation", "true");
        sparkProperties.put("spark.jars", jarPath);

        sparkSubmitData.put("sparkProperties", sparkProperties);

        JSONArray appArgs = new JSONArray();
        appArgs.add(job.getId());
        sparkSubmitData.put("appArgs", appArgs);
        Map<String, String> header = new HashMap<>();
        header.put("Content-Type", "application/json;charset=UTF-8");

        System.out.println(sparkSubmitData);

        String resultStr = this.httpClient.post("v1/submissions/create", sparkSubmitData, this.sparkServer, this.sparkRestfulPort, header,
                Long.valueOf(3000L));
        JSONObject result = JSON.parseObject(resultStr);

        Boolean status = result.getBoolean("success");
        if (!status.booleanValue()) {
            throw new Exception(result.getString("message"));
        }
        job.setStatus("RUNNING");
        this.jobRepository.save(job);
    }


    public synchronized Boolean checkSparkServerResource(Integer cores, Long memoryGB) throws Exception {
        String resultStr = this.httpClient.get("json", this.sparkServer, this.sparkPort, null);
        JSONObject sparkResource = JSON.parseObject(resultStr);
        Integer totalCores = sparkResource.getInteger("cores");
        Integer userdCores = sparkResource.getInteger("coresused");

        Long totalMemory = sparkResource.getLong("memory");
        Long usedMemory = sparkResource.getLong("memoryused");

        Integer freeCores = Integer.valueOf(totalCores.intValue() - userdCores.intValue() - cores.intValue() / 2);
        Long freeMemory = Long.valueOf(totalMemory.longValue() - usedMemory.longValue() - (cores.intValue() / 2));

        Boolean enough = Boolean.valueOf(true);
        if (cores.intValue() > freeCores.intValue() || memoryGB.longValue() * 1024L > freeMemory.longValue()) {
            enough = Boolean.valueOf(false);
        }
        return enough;
    }


    public void deleteJob(String jobId) {
        /* 384 */
        JobEntity job = this.jobRepository.findById(jobId).get();
        /* 385 */
        String modelId = job.getModelId();

        /* 387 */
        DeletedJobEntity deletedJobEntity = new DeletedJobEntity();
        /* 388 */
        deletedJobEntity.setId(Uid.generateShortUuid());
        /* 389 */
        deletedJobEntity.setJobId(jobId);
        /* 390 */
        deletedJobEntity.setDeleteTime(new Date());
        /* 391 */
        deletedJobEntity.setModelId(modelId);
        /* 392 */
        this.deletedJobRepository.save(deletedJobEntity);

        /* 394 */
        this.jobRepository.deleteById(jobId);
    }


    public void deleteJobs(List<JobEntity> jobEntities) {
        /* 402 */
        List<DeletedJobEntity> deletedJobEntities = new ArrayList<>();
        /* 403 */
        Date deleteTime = new Date();
        /* 404 */
        for (JobEntity job : jobEntities) {
            /* 405 */
            String jobId = job.getId();

            /* 407 */
            DeletedJobEntity deletedJob = new DeletedJobEntity();
            /* 408 */
            deletedJob.setId(Uid.generateShortUuid());
            /* 409 */
            deletedJob.setDeleteTime(deleteTime);
            /* 410 */
            deletedJob.setJobId(jobId);
            /* 411 */
            deletedJobEntities.add(deletedJob);
        }
        /* 413 */
        this.deletedJobRepository.saveAll(deletedJobEntities);
        /* 414 */
        this.jobRepository.deleteAll(jobEntities);
    }

    public JobEntity jobEncode(JobEntity job) throws Exception {
        /* 418 */
        if (StringUtil.isNotEmpty(job.getParam())) {
            /* 419 */
            job.setParam(URLEncoder.encode(job.getParam(), "UTF-8"));
        }
        /* 421 */
        return job;
    }
}


/* Location:              C:\Users\admin\Desktop\xxm\zjg11\ZjgCaculate-v1.0.jar!\BOOT-INF\classes\cn\org\intelli\zffserver\service\JobService.class
 * Java compiler version: 8 (52.0)
 * JD-Core Version:       1.1.3
 */